﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Diagnostics;

namespace Agent.Messaging
{
  /// <summary>
  /// Class for sending messages to the <see cref="MessageReceiver"/> class, typically across 
  /// different processes.
  /// </summary>
  /// <remarks>
  /// This class is thread safe
  /// TODO: This class (and the receiver) should ideally be able to send messages of type T, where
  /// T is any kind of data (not just a string)
  /// </remarks>
  public class MessageSender : IDisposable
  {
    #region fields
    const int INITIAL_TIMEOUT_MS = 1000;
    const int REACQUIRE_TIMEOUT_MS = 1000;
    const int WAIT_TO_REACQUIRE_MUTEX_MS = 0;

    StateMachineHelper<SenderReceiverStateMachineState> stateMachine;

    Queue<MessageWrapper> queuedMessages;

    MessageQueue queue;
    Mutex channelMutex;
    string channelName;

    int instanceId;
    static int nextThreadId = 0;
    static int nextInstanceId = 0;

    static object staticLockObject = typeof(MessageSender);

    AutoResetEvent messagesToBeSent = null;
    ManualResetEvent exitSendLoopSignal = null;
    ManualResetEvent sendLoopTerminatedSignal = null;
    Thread sendingThread = null;
    DetailedLogger logger;

    SenderReceiverStateMachineState sendThreadExitState;

    static Dictionary<SenderReceiverStateMachineState, IEnumerable<SenderReceiverStateMachineState>> allowedTransitions;
    static IEnumerable<SenderReceiverStateMachineState> sinkStates;
    #endregion

    /// <summary>
    /// Sets up the state machine transitions
    /// </summary>
    static MessageSender()
    {
      allowedTransitions = new Dictionary<SenderReceiverStateMachineState, IEnumerable<SenderReceiverStateMachineState>>();
      allowedTransitions[SenderReceiverStateMachineState.Uninitialized] = new[] { SenderReceiverStateMachineState.Initialized };
      allowedTransitions[SenderReceiverStateMachineState.Initialized] = new[] { SenderReceiverStateMachineState.Starting };
      allowedTransitions[SenderReceiverStateMachineState.Starting] = new[] { SenderReceiverStateMachineState.Running };
      allowedTransitions[SenderReceiverStateMachineState.Running] = new SenderReceiverStateMachineState[] { };
      allowedTransitions[SenderReceiverStateMachineState.Stopping] = new[] { SenderReceiverStateMachineState.Initialized };

      sinkStates = new[] { SenderReceiverStateMachineState.Stopping, SenderReceiverStateMachineState.Disposed, SenderReceiverStateMachineState.Error };
    }

    /// <summary>
    /// Creates a new MessageSender object
    /// </summary>
    /// <param name="channelName">The name of the message channel</param>
    /// <remarks>
    /// This name must be the same name that is used by the <see cref="MessageReceiver"/> class
    /// </remarks>
    public MessageSender(string channelName)
    {
      lock (staticLockObject)
      {
        instanceId = nextInstanceId++;
      }

      stateMachine = new StateMachineHelper<SenderReceiverStateMachineState>(allowedTransitions, sinkStates, SenderReceiverStateMachineState.Uninitialized, true);

      this.channelName = channelName;
      logger = new DetailedLogger(channelName + "_" + instanceId);
      logger.Log("Starting up... ");

      try
      {
        channelMutex = new Mutex(false, channelName + MessageReceiver.MUTEX_SUFFIX);
        queue = MessageQueue.OpenQueue(channelName + MessageReceiver.QUEUE_SUFFIX);
        messagesToBeSent = new AutoResetEvent(false);
        exitSendLoopSignal = new ManualResetEvent(false);
        sendLoopTerminatedSignal = new ManualResetEvent(false);
        queuedMessages = new Queue<MessageWrapper>();
      }
      catch
      {
        logger.Log("Hit exception during construction... cleaning up", LoggerSeverity.Error);
        CleanUpResources();
        stateMachine.GoToState(SenderReceiverStateMachineState.Error);
        throw;
      }

      stateMachine.GoToState(SenderReceiverStateMachineState.Initialized);
      logger.Log("Constructed OK");
    }

    /// <summary>
    /// Cleans up the resources in the case of error / shutdown
    /// </summary>
    private void CleanUpResources()
    {
      if (logger != null && stateMachine.IsInState(SenderReceiverStateMachineState.Disposed) != true)
        logger.Log("Cleaning up...");

      if (channelMutex != null)
        channelMutex.Dispose();
      channelMutex = null;

      if (queue != null)
        queue.Dispose();
      queue = null;

      if (messagesToBeSent != null)
        messagesToBeSent.Dispose();
      messagesToBeSent = null;

      if (exitSendLoopSignal != null)
        exitSendLoopSignal.Dispose();
      exitSendLoopSignal = null;

      if (sendLoopTerminatedSignal != null)
        sendLoopTerminatedSignal.Dispose();
      sendLoopTerminatedSignal = null;

      queuedMessages = null;
    }

    /// <summary>
    /// Starts the sender, making it possible to send messages via the <see cref="SendMessage" /> method
    /// </summary>
    public void Start()
    {
      logger.Log("Starting; current state is " + stateMachine.CurrentState);

      lock (this)
      {
        // OK to call start multiple times; it's a no-op. Also gate with CanGoToState in case we change the allowed transitions
        if (stateMachine.CanGoToState(SenderReceiverStateMachineState.Starting) && stateMachine.IsInState(new[] { SenderReceiverStateMachineState.Starting, SenderReceiverStateMachineState.Running }))
          return;

        stateMachine.GoToState(SenderReceiverStateMachineState.Starting);
      }

      if (sendingThread == null || sendingThread.IsAlive == false)
      {
        logger.Log("Creating main send thread object");
        sendingThread = new Thread(MessageSendLoop);
      }

      logger.Log("Kicking off the main send thread");
      exitSendLoopSignal.Reset();
      sendLoopTerminatedSignal.Reset();
      sendThreadExitState = SenderReceiverStateMachineState.Starting;
      sendingThread.Start();
      new Thread(SendThreadExitHandler).Start();
    }

    /// <summary>
    /// Method that determines what to do when the main sending loop terminates
    /// </summary>
    /// <param name="state">Dummy parameter</param>
    /// <remarks>
    /// This method waits for the thread running <see cref="MessageSendLoop" /> to end, then decides
    /// which state to transition to based on the exit status of that thread
    /// </remarks>
    void SendThreadExitHandler(object state)
    {
      lock (staticLockObject)
      {
        Thread.CurrentThread.Name = "loop exit handler " + nextThreadId++;
      }

      logger.Log("Waiting for main wait loop to terminate... joining " + sendingThread.Name);
      sendingThread.Join();
      logger.Log("Joined! Current state is " + stateMachine.CurrentState + " and thread exit state was " + sendThreadExitState);

      lock (this)
      {
        switch (sendThreadExitState)
        {
          case SenderReceiverStateMachineState.Error:
            logger.Log("Thread ended with an error, going to error state", LoggerSeverity.Warning);
            stateMachine.GoToState(SenderReceiverStateMachineState.Error);
            break;

          case SenderReceiverStateMachineState.Stopping:
            logger.Log("Thread ended because it was asked to stop, going to initialized state", LoggerSeverity.Trace);
            stateMachine.GoToState(SenderReceiverStateMachineState.Initialized);
            break;

          default:
            logger.Log("Unknown thread exit value " + sendThreadExitState, LoggerSeverity.Error);
            Debug.Assert(false, "Unknown thread exit value " + sendThreadExitState);
            break;
        }
      }

      // Call user code outside of the lock (above)
      switch (sendThreadExitState)
      {
        case SenderReceiverStateMachineState.Error:
          RaiseFailed();
          break;

        case SenderReceiverStateMachineState.Stopping:
          RaiseStopped();
          break;
      }

      sendLoopTerminatedSignal.Set();
      logger.Log("Thread is exiting");
    }

    /// <summary>
    /// Signals the sender thread to stop, but returns immediately
    /// </summary>
    /// <remarks>
    /// The sending thread will finish asynchronously
    /// </remarks>
    public void StopAsync()
    {
      Stop(false);
    }

    /// <summary>
    /// Signals the sending thread to stop, and waits for it to complete
    /// </summary>
    public void Stop()
    {
      Stop(true);
    }

    /// <summary>
    /// Signals the sending thread to stop, optionally waiting for it to complete
    /// </summary>
    /// <param name="waitUntilStopped">true to wait for the thread to exit; false otherwise</param>
    void Stop(bool waitUntilStopped)
    {
      logger.Log("Stopping... wait=" + waitUntilStopped);

      if (waitUntilStopped && Thread.CurrentThread == sendingThread)
      {
        logger.Log("Can't call from the main loop thread; it might deadlock", LoggerSeverity.Error);
        Debug.Assert(false, "Can't call Stop from the main loop thread; will deadlock");
      }

      if (waitUntilStopped)
        StopImpl();
      else
        ThreadPool.QueueUserWorkItem(delegate { StopImpl(); });

    }

    /// <summary>
    /// Actually implements the Stop procedure; this method blocks
    /// </summary>
    void StopImpl()
    {
      // Give the state machine a chance to start running
      while (CurrentState == SenderReceiverStateMachineState.Starting)
        Thread.Sleep(1);

      lock (this)
      {
        // All these states are no-ops; also gate with CanGoToState in case we change the rules
        if (stateMachine.CanGoToState(SenderReceiverStateMachineState.Stopping) && stateMachine.IsInState(new[] { SenderReceiverStateMachineState.Stopping, SenderReceiverStateMachineState.Uninitialized, SenderReceiverStateMachineState.Initialized }))
          return;

        if (sendingThread.IsAlive)
          stateMachine.GoToState(SenderReceiverStateMachineState.Stopping);
      }

      logger.Log("Sending STOP message to the sending thread; it should terminate itself");
      SendMessage(new MessageWrapper(MessageType.StopMessage));

      logger.Log("Waiting until the thread stopped event is set...");
      sendLoopTerminatedSignal.WaitOne();
      logger.Log(".. and done waiting!");
    }

    /// <summary>
    /// The main sending loop
    /// </summary>
    /// <remarks>
    /// This method runs an infinite loop, waiting for messages to be posted via the <see cref="SendMessage"/>
    /// method. The loop terminates if the object is being disposed, if someone calls <see cref="Stop"/>, or if 
    /// there is an internal error
    /// </remarks>
    void MessageSendLoop()
    {
      lock (staticLockObject)
      {
        Thread.CurrentThread.Name = "send thread " + nextThreadId++;
      }

      stateMachine.GoToState(SenderReceiverStateMachineState.Running);
      sendThreadExitState = SenderReceiverStateMachineState.Error;

      logger.Log("About to wait on mutex with an initial timeout");
      WaitHandleResult result = channelMutex.SafeWaitOne(INITIAL_TIMEOUT_MS);
      if (result != WaitHandleResult.Acquired)
      {
        logger.Log("Can't acquire own mutex; bailing... ", LoggerSeverity.Error);
        return;
      }

      logger.Log("About to start the main loop");

      while (true)
      {
        logger.Log("About to wait indefinitely for the pending message flag");
        result = messagesToBeSent.SafeWaitOne(Timeout.Infinite);
        if (result != WaitHandleResult.Acquired)
        {
          logger.Log("Couldn't acquire handle; bailing", LoggerSeverity.Error);
          break;
        }

        logger.Log("There are one or more messages in the queue", LoggerSeverity.Information);

        MessageWrapper[] messages;

        lock (this)
        {
          logger.Log("There are " + queuedMessages.Count + " messages pending; making local copy");
          messages = new MessageWrapper[queuedMessages.Count];
          queuedMessages.CopyTo(messages, 0);
          queuedMessages.Clear();
        }

        foreach (MessageWrapper wrapper in messages)
        {
          if (wrapper.MessageType == MessageType.StopMessage)
          {
            logger.Log("Processed STOP message; going to wind down the send loop", LoggerSeverity.Trace);
            sendThreadExitState = SenderReceiverStateMachineState.Stopping;
            break;
          }
          string message = wrapper.UserMessage;
          logger.Log("About to send message '" + message + "'", LoggerSeverity.Information);
          queue.EnqueueMessage(message);
        }

        logger.Log("Releasing mutex to signal receiver that we have sent some messages...");
        channelMutex.ReleaseMutex();

        if (sendThreadExitState == SenderReceiverStateMachineState.Stopping)
        {
          logger.Log("Breaking from main loop because we've been asked to stop");
          break;
        }

        logger.Log("Sleep() to let someone else take the mutex, if necessary Zzzz...");
        Thread.Sleep(WAIT_TO_REACQUIRE_MUTEX_MS);

        logger.Log("Zzzz waking up, taking the mutex again");
        result = channelMutex.SafeWaitOne(REACQUIRE_TIMEOUT_MS);
        if (result != WaitHandleResult.Acquired)
        {
          logger.Log("Can't re-acquire mutex fast enough; bailing...", LoggerSeverity.Error);
          sendThreadExitState = SenderReceiverStateMachineState.Error;
          break;
        }
      }

      logger.Log("Thread is exiting with code " + sendThreadExitState);
    }

    /// <summary>
    /// Queues up a message to be sent
    /// </summary>
    /// <param name="messageWrapper">The message to send</param>
    /// <remarks>
    /// All messages are sent asynchronously
    /// </remarks>
    void SendMessage(MessageWrapper messageWrapper)
    {
      logger.Log("About to queue up message of type " + messageWrapper.MessageType + " with message '" + (messageWrapper.UserMessage ?? "<null>") + "'", LoggerSeverity.Information);

      lock (this)
      {
        queuedMessages.Enqueue(messageWrapper);
      }

      logger.Log("About to set the event to awaken the send thread", LoggerSeverity.Information);
      messagesToBeSent.Set();
    }

    /// <summary>
    /// Queues up a message to be sent to the receiver
    /// </summary>
    /// <param name="message">The message to send</param>
    /// <remarks>
    /// All messages are sent asynchronously
    /// </remarks>
    public void SendMessage(string message)
    {
      logger.Log("About to queue up message '" + message + "'", LoggerSeverity.Information);

      if (stateMachine.IsInState(new SenderReceiverStateMachineState[] { SenderReceiverStateMachineState.Starting, SenderReceiverStateMachineState.Running} ) != true)
        throw new InvalidOperationException("Sender is not currently running");

      SendMessage(new MessageWrapper(MessageType.UserMessage, message));
    }

    /// <summary>
    /// Disposes of the object, releasing any resources
    /// </summary>
    public void Dispose()
    {
      Dispose(false);
      GC.SuppressFinalize(this);
    }

    /// <summary>
    /// Disposes of the object, releasing any resources
    /// </summary>
    /// <param name="isFinalizing">Whether the object is being finalized</param>
    void Dispose(bool isFinalizing)
    {
      logger.Log("isFinalizing = " + isFinalizing + ",  currentState = " + stateMachine.CurrentState);
      if (isFinalizing)
        logger.Log("Exceptions are expected if some items are already disposed...");

      lock (this)
      {
        if (stateMachine.CurrentState == SenderReceiverStateMachineState.Disposed)
        {
          logger.Log("Already disposing or disposed; returning immediately");
          return;
        }

        logger.Log("Going to disposed state");
        stateMachine.GoToState(SenderReceiverStateMachineState.Disposed);
      }

      try
      {
        Stop();
      }
      catch (ObjectDisposedException)
      {
        logger.Log("Exception trying to set the stop listening event because it is already disposed",
          isFinalizing ? LoggerSeverity.Information : LoggerSeverity.Error);
      }

      CleanUpResources();
    }

    /// <summary>
    /// Finalizer that calls <see cref="Dispose"/>
    /// </summary>
    ~MessageSender()
    {
      logger.Log("Finalizing...");
      Dispose(true);
    }

    /// <summary>
    /// Helper to raise the <see cref="Failed"/> event
    /// </summary>
    void RaiseFailed()
    {
      logger.Log("Raising the Failed event; current state = " + stateMachine.CurrentState);
      var handler = Failed;
      if (handler != null)
        handler(this, EventArgs.Empty);
    }

    /// <summary>
    /// Helper to raise the <see cref="Stopped"/> event
    /// </summary>
    void RaiseStopped()
    {
      logger.Log("Raising the Stopped event; current state = " + stateMachine.CurrentState);
      var handler = Stopped;
      if (handler != null)
        handler(this, EventArgs.Empty);
    }

    /// <summary>
    /// Raised when the receiver fails for any reason (eg, can't acquire system resources)
    /// </summary>
    /// <remarks>
    /// This event might be raised on a background thread
    /// </remarks>
    public event EventHandler Failed;

    /// <summary>
    /// Raised when the receiver has successfully stopped
    /// </summary>
    /// <remarks>
    /// This event might be raised on a background thread
    /// </remarks>
    public event EventHandler Stopped;

    /// <summary>
    /// Returns the current state of the state machine
    /// </summary>
    public SenderReceiverStateMachineState CurrentState { get { return stateMachine.CurrentState; } }
  }
}
